package com.jblv.frame.common.kafka;

import com.jblv.frame.common.kafka.bean.MqBaseMessage;
import com.jblv.frame.common.kafka.config.KafkaConfigUtils;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author happyonion.yangtao
 * @version 1.0
 * @title KafkaProducerUtils
 * @projectName jblv-frame
 * @packageName com.jblv.web.kafka
 * @description Kafka消息发送工具类
 * @date Create at 2020/4/17 13:52
 * @since jdk1.8
 */
public class KafkaProducerUtils {
    private final AtomicLong value = new AtomicLong(0L);
    private final Map<String, Producer<String, MqBaseMessage>> producerMap = new HashMap();
    private static KafkaProducerUtils utils = null;

    public Producer<String, MqBaseMessage> getProducer(String topic) {
        Producer<String, MqBaseMessage> producer = null;
        if (producerMap.containsKey(topic)) {
            producer = producerMap.get(topic);
        }

        if (producer == null) {
            synchronized(producerMap) {
                if (producerMap.containsKey(topic)) {
                    return producerMap.get(topic);
                }

                ProducerConfig producerConfig =  KafkaConfigUtils.getProducerConfig();
                producer = new Producer(producerConfig);
                producerMap.put(topic, producer);
            }
        }

        return producer;
    }

    private KafkaProducerUtils() {
    }

    public static KafkaProducerUtils newInstance() {
        if (utils == null) {
            synchronized(KafkaProducerUtils.class) {
                if (utils == null) {
                    utils = new KafkaProducerUtils();
                }
            }
        }

        return utils;
    }

    public void close() {
        Iterator var1 = this.producerMap.entrySet().iterator();

        while(var1.hasNext()) {
            Map.Entry<String, Producer<String, MqBaseMessage>> entry = (Map.Entry)var1.next();
            (entry.getValue()).close();
        }

    }

    public boolean send(Object obj, String topic) {
        try {
            MqBaseMessage bkm = new MqBaseMessage();
            bkm.setId(1L);
            bkm.setObj(obj);
            KeyedMessage<String, MqBaseMessage> km = new KeyedMessage(topic, topic + "." + this.value.getAndIncrement(), bkm);
            utils.getProducer(topic).send(km);
            return true;
        } catch (Exception var5) {
            var5.printStackTrace();
            return false;
        }
    }
}
